Setup¶
In [1]:
import pyspark
from pyspark.sql import SparkSession, Row
import pyspark.sql.functions as F
import pyspark.sql.types as T
In [2]:
import numpy as np
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt
import matplotlib.ticker as tck
import statistics as stats
%matplotlib inline
In [3]:
spark = SparkSession.builder.appName("WeatherData_Analysis").getOrCreate()
In [4]:
spark
Out[4]:
Load Data¶
In [5]:
#define schema
schema = T.StructType([
T.StructField("Date",T.DateType()),
T.StructField("MinTemperature",T.DoubleType()),
T.StructField("MaxTemperature",T.DoubleType()),
T.StructField("Precipitation",T.DoubleType()),
T.StructField("RelativeHumidity",T.DoubleType()),
T.StructField("State",T.StringType()),
T.StructField("District",T.StringType()),
])
In [6]:
#load data
weatherData = spark.read.load("Datasets/Output/WeatherDistrictData.csv",
format="csv",schema=schema,
header=True)
In [7]:
#check schema
weatherData.printSchema()
In [8]:
#adding additional columns
# Month: for aggregating at month level
# District_State : for agregating at district level
weatherData = (weatherData
.withColumn("Month",F.concat_ws("-",F.year("Date"),
F.when(F.month("Date").between(1,9),F.concat(F.lit("0"),F.month("Date")))
.otherwise(F.month("Date"))))
.withColumn("District_State", F.concat_ws(", ","District","State"))
)
In [9]:
#Check Schema
weatherData.printSchema()
In [10]:
#persist
weatherData.persist()
Out[10]:
In [11]:
weatherData.rdd.getNumPartitions()
Out[11]:
In [12]:
#Check Plan
weatherData.explain()
In [13]:
#check number of rows
#we have data for 2000-2013 i.e. 5114 days
# and 535 districts
#We should have 2735990 rows
weatherData.count()
Out[13]:
In [14]:
#check data
weatherData.show()
Generate list of States¶
In [15]:
states = list(sorted([row["State"] for row in weatherData.select("State").distinct().collect()]))
Plotting¶
In [16]:
def genBarPlot(df,state,column):
plt.figure(figsize=(30,12))
plt.bar(df.Month,df[column], align='edge', width=0.4)
plt.xlabel("Time",size=16)
plt.ylabel(column,size=16)
plt.xticks(ticks=np.arange(0,5113,3),rotation=90,size=16)
plt.yticks(size=16)
plt.xlim([0,len(df.Month)])
plt.title(state+" Monthly "+column,size=20)
plt.show()
In [17]:
def genDualLinePlot(df,state,column1,column2,yaxis="",color1="red",color2="blue"):
plt.figure(figsize=(30,12))
plt.plot(df.Month,df[column1],color=color1,label=column1)
plt.plot(df.Month,df[column2],color=color2,label=column2)
plt.xlabel("Time",size=16)
plt.ylabel(yaxis,size=16)
plt.xticks(ticks=np.arange(0,5113,3),rotation=90,size=16)
plt.yticks(size=16)
plt.xlim([0,len(df.Month)])
plt.title(state+" Monthly "+column1+" and "+column2,size=20)
plt.show()
In [18]:
def genPlots(dataframe,state=None):
if state:
df = dataframe.filter(F.col("State")==state).toPandas().sort_values("Month")
else:
df = dataframe.toPandas().sort_values("Month")
state = "India"
print("Weather Information for "+state)
print(df.describe())
genBarPlot(df,state,"TotalPrecipitation")
genBarPlot(df,state,"AverageRelativeHumidity")
genDualLinePlot(df,state,"AverageMaxTemperature","AverageMinTemperature",yaxis="Temperature")
Aggregate for India¶
In [19]:
#CReate GroupedDataFrame
monthlyDF = weatherData.groupBy("Month")
In [20]:
#Create Aggregations and Rename Columns
monthlyDF = (monthlyDF.agg(F.sum("Precipitation"),F.mean("MinTemperature"),F.mean("MaxTemperature"),F.mean("RelativeHumidity"))
.select("Month",
F.col("sum(Precipitation)").alias("TotalPrecipitation"),
F.col("avg(MinTemperature)").alias("AverageMinTemperature"),
F.col("avg(MaxTemperature)").alias("AverageMaxTemperature"),
F.col("avg(RelativeHumidity)").alias("AverageRelativeHumidity"))
)
In [21]:
#Generate Plots
genPlots(monthlyDF)
Findings:¶
- Precipitation:
- The Total Preciptation recieved by India peaks in the months June-September. This is the result of the monsoon.
- The year 2002 recieved comparatively less rainfall during the monsoon period than its preceding years, with the exception of August.
- The year 2003 saw exceptionally high amount of precipitation during the monsoon, with July having a recond rainfall of over 275000 mm (averaging to over 500 mm in 535 districts).
- The year 2005 again saw comparatively less rainfall, with the exception of July which remained close to the normal expectation.
- There is a slight increase in total precipitation from 2007 onwards during the monsoon period.
- The years 2011-2013 saw a slight increase over the normal observations for 2007-2010.
- Relative Humidity:
- India observes a fairly high mean relative humidity, with most months having mean relative humidity 0.4 to 0.8.
- The months of Feburary, March and April are exception to this with values ranging between 0.3 to 0.5.
- The Mean Relative Humidy peaks during the monsoon season in which it ranges between 0.6 to 0.8.
- There also appears to be a upward trend in the mean humidity value over the years.
- Temperatures:
- India records a wairly wide gap in mean minimum and maximum temperatures with a approximate difference of 15 Celcius.
- The national average for minimum temperature ranges between 10-15ºC during the winter months(October to late Feburary), and 20-27ºC during the summer months(May to late August).
- The national average for maximum temperature ranges between 27-30ºC during the winter months(October to late Feburary) post which there is a sudden increse in maximum temperature with early April recording temperatures above 35ºC.
- The maximum temperature then ranges between 35-40ºC until the monsoon sets in during which the temperatures go down to approx. 32ºC. There is a second peak in maximum temperature in late September to early October at the end of monsoon.
- While the annual trend for mean minimum temperature remains close to normal with smooth corves, the mean maximum temperature records a more irregular behaviour in later years with more pointed turns and sharp peaks.
Aggregating data for each State¶
Aggregated at Month Level
In [22]:
#Create GroupedDataFrame
monthlyStateDF = weatherData.groupBy(F.col("State"),F.col("Month"))
Aggregations used:
- Precipitation : sum : To get total precipitation(rainfall) recieved by the state
- MinTemperature : mean : To get average minimum temperature for the state
- MaxTemperature : mean : To get average maximum temperature for the state
- RelativeHumidity : mean : To get average relative humidity for the state
In [23]:
#Apply Aggregations and rename columns
monthlyStateDF = (monthlyStateDF.agg(F.sum("Precipitation"),F.mean("MinTemperature"),F.mean("MaxTemperature"),F.mean("RelativeHumidity"))
.select("State","Month",
F.col("sum(Precipitation)").alias("TotalPrecipitation"),
F.col("avg(MinTemperature)").alias("AverageMinTemperature"),
F.col("avg(MaxTemperature)").alias("AverageMaxTemperature"),
F.col("avg(RelativeHumidity)").alias("AverageRelativeHumidity"))
)
In [24]:
#Check Schema
monthlyStateDF.printSchema()
In [25]:
#check Execution Plan
monthlyStateDF.persist()
monthlyStateDF.explain()
In [26]:
for state in states:
genPlots(monthlyStateDF,state)
Top 10 Districts Ranking¶
Aggregated at montly level
Aggregations used:
- Precipitation : sum : To get total precipitation(rainfall) recieved by the district
- MinTemperature : min : To get minimum temperature recorded by the district
- MaxTemperature : min : To get maximum temperature recorded by the district
- RelativeHumidity : min : To get minimum relative humidity recorded by the district
- RelativeHumidity : max : To get maximum relative humidity recorded by the district
In [27]:
#Create GroupedDataFrame and Aggregations
districtsDF = (weatherData.groupBy("Month","District_State")
.agg(F.sum("Precipitation"),F.min("MinTemperature"),F.max("MaxTemperature"),F.min("RelativeHumidity"),F.max("RelativeHumidity"))
.select("Month","District_State",
F.col("sum(Precipitation)").alias("TotalPrecipitation"),
F.col("min(MinTemperature)").alias("MinTemperature"),
F.col("max(MaxTemperature)").alias("MaxTemperature"),
F.col("max(RelativeHumidity)").alias("MaximumRelativeHumidity"),
F.col("min(RelativeHumidity)").alias("MinimumRelativeHumidity"))
)
In [28]:
districtsDF.persist()
Out[28]:
Hottest Districts¶
In [29]:
hottestDistricts = districtsDF.select("Month","District_State","MaxTemperature")
In [30]:
@F.pandas_udf("Month string,District_State string,MaxTemperature double",F.PandasUDFType.GROUPED_MAP)
def get10MaxTemp(pdf):
return pdf.nlargest(10,"MaxTemperature")
In [31]:
hottestDistricts = hottestDistricts.groupBy("Month").apply(get10MaxTemp)
In [32]:
hottestDistricts
Out[32]:
Coldest Districts¶
In [33]:
coldestDistricts = districtsDF.select("Month","District_State","MinTemperature")
In [34]:
@F.pandas_udf("Month string,District_State string,MinTemperature double",F.PandasUDFType.GROUPED_MAP)
def get10MinTemp(pdf):
return pdf.nsmallest(10,"MinTemperature")
In [35]:
coldestDistricts = coldestDistricts.groupBy("Month").apply(get10MinTemp)
In [36]:
coldestDistricts
Out[36]:
Districts with Maximum Precipitation¶
In [37]:
wettestDistricts = districtsDF.select("Month","District_State","TotalPrecipitation")
In [38]:
@F.pandas_udf("Month string,District_State string,TotalPrecipitation double",F.PandasUDFType.GROUPED_MAP)
def get10MaxPreci(pdf):
return pdf.nlargest(10,"TotalPrecipitation")
In [39]:
wettestDistricts = wettestDistricts.groupBy("Month").apply(get10MaxPreci)
In [40]:
wettestDistricts
Out[40]:
Districts with Minimum Precipitation¶
In [41]:
driestDistricts = districtsDF.select("Month","District_State","TotalPrecipitation")
In [42]:
@F.pandas_udf("Month string,District_State string,TotalPrecipitation double",F.PandasUDFType.GROUPED_MAP)
def get10MinPreci(pdf):
return pdf.nsmallest(10,"TotalPrecipitation")
In [43]:
driestDistricts = driestDistricts.groupBy("Month").apply(get10MinPreci)
In [44]:
driestDistricts
Out[44]:
Districts with Maximum Humidity¶
In [45]:
mostHumidDistricts = districtsDF.select("Month","District_State","MaximumRelativeHumidity")
In [46]:
@F.pandas_udf("Month string,District_State string,MaximumRelativeHumidity double",F.PandasUDFType.GROUPED_MAP)
def get10MaxHumid(pdf):
return pdf.nlargest(10,"MaximumRelativeHumidity")
In [47]:
mostHumidDistricts = mostHumidDistricts.groupBy("Month").apply(get10MaxHumid)
In [48]:
mostHumidDistricts
Out[48]:
Districts with Minimum Humidity¶
In [49]:
leastHumidDistricts = districtsDF.select("Month","District_State","MinimumRelativeHumidity")
In [50]:
@F.pandas_udf("Month string,District_State string,MinimumRelativeHumidity double",F.PandasUDFType.GROUPED_MAP)
def get10MinHumid(pdf):
return pdf.nsmallest(10,"MinimumRelativeHumidity")
In [51]:
leastHumidDistricts = leastHumidDistricts.groupBy("Month").apply(get10MinHumid)
In [52]:
leastHumidDistricts
Out[52]:
Storing Results to FileSystem for Tableau Vizualization¶
Tableau Viz¶
Link: Tableav Viz
In [53]:
%%HTML
<div class='tableauPlaceholder' id='viz1601471864533' style='position: relative'><noscript><a href='#'><img alt=' ' src='https://public.tableau.com/static/images/We/WeatherAnalysis_16014109382880/IndianWeatherOvertheYears/1_rss.png' style='border: none' /></a></noscript><object class='tableauViz' style='display:none;'><param name='host_url' value='https%3A%2F%2Fpublic.tableau.com%2F' /> <param name='embed_code_version' value='3' /> <param name='site_root' value='' /><param name='name' value='WeatherAnalysis_16014109382880/IndianWeatherOvertheYears' /><param name='tabs' value='yes' /><param name='toolbar' value='yes' /><param name='static_image' value='https://public.tableau.com/static/images/We/WeatherAnalysis_16014109382880/IndianWeatherOvertheYears/1.png' /> <param name='animate_transition' value='yes' /><param name='display_static_image' value='yes' /><param name='display_spinner' value='yes' /><param name='display_overlay' value='yes' /><param name='display_count' value='yes' /><param name='language' value='en' /><param name='filter' value='publish=yes' /></object></div> <script type='text/javascript'> var divElement = document.getElementById('viz1601471864533'); var vizElement = divElement.getElementsByTagName('object')[0]; if ( divElement.offsetWidth > 800 ) { vizElement.style.width='900px';vizElement.style.height='650px';} else if ( divElement.offsetWidth > 500 ) { vizElement.style.width='900px';vizElement.style.height='650px';} else { vizElement.style.minWidth='900px';vizElement.style.maxWidth='100%';vizElement.style.height='1050px';} var scriptElement = document.createElement('script'); scriptElement.src = 'https://public.tableau.com/javascripts/api/viz_v1.js'; vizElement.parentNode.insertBefore(scriptElement, vizElement); </script>
